package com.jscloud.bigdata.mr.opt2;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class GroupMain extends Configured implements Tool {
        @Override
        public int run(String[] args) throws Exception {
                //获取job对象
                Job job = Job.getInstance(super.getConf(), "group");
                //第一步：读取文件，解析成为key，value对
                job.setInputFormatClass(TextInputFormat.class);
                TextInputFormat.addInputPath(job, new Path("file:///D:\\JSProjects\\jsCloud-bigdata-app\\mrapp\\input2"));

                //第二步：自定义map逻辑
                job.setMapperClass(GroupMapper.class);
                job.setMapOutputKeyClass(OrderBean.class);
                job.setMapOutputValueClass(NullWritable.class);

                //第三步：分区
                job.setPartitionerClass(GroupPartition.class);

                //第四步：排序  已经做了

                //第五步：规约  combiner  省掉

                //第六步：分组   自定义分组逻辑
                job.setGroupingComparatorClass(MyGroup.class);

                //第七步：设置reduce逻辑
                job.setReducerClass(GroupReducer.class);
                job.setOutputKeyClass(OrderBean.class);
                job.setOutputValueClass(NullWritable.class);

                //第八步：设置输出路径
                job.setOutputFormatClass(TextOutputFormat.class);
                TextOutputFormat.setOutputPath(job, new Path("file:///D:\\JSProjects\\jsCloud-bigdata-app\\mrapp\\output2"));

                boolean b = job.waitForCompletion(true);
                return b ? 0 : 1;
        }

        public static void main(String[] args) throws Exception {
                int run = ToolRunner.run(new Configuration(), new GroupMain(), args);
                System.exit(run);
        }

}
